package com.chenye.storm.test;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.Utils;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import static com.chenye.storm.test.RocketMqConfig.CONSUMER_GROUP;
import static com.chenye.storm.test.RocketMqConfig.CONSUMER_TOPIC;


/**
 * @Auther:chenye
 * @Data:2018/6/11
 * @Description:
 */
public class Topology {

    public static void main( String[] args ) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        TopologyBuilder builder = new TopologyBuilder();
        Properties properties = new Properties();
        properties.setProperty(SpoutConfig.NAME_SERVER_ADDR, "127.0.0.1:9876");
        properties.setProperty(CONSUMER_GROUP, "storm-consumer");
        properties.setProperty(CONSUMER_TOPIC, "TopicTest");

        RocketMqSpout spout = new RocketMqSpout(properties);

        builder.setSpout("rocketMqSpout",spout,1).setNumTasks(2);
        builder.setBolt("testBolt",new TestBolt(),1).shuffleGrouping("rocketMqSpout").setNumTasks(2);
        builder.setBolt("testBolt1",new TestBolt1(),1).shuffleGrouping("rocketMqSpout").setNumTasks(2);
        //本地提交
        LocalCluster cluster = new LocalCluster();
        Config config = new Config();
//        config.setNumAckers(0);//相当于关闭ack确认，因为在spout将消息发送出去后，就会触发ack
        cluster.submitTopology("topologyName", config, builder.createTopology());

        Utils.sleep(10000);
//        cluster.killTopology("topologyName");
//        cluster.shutdown();
    }
}
